"""
FFmpeg服务

提供FFmpeg相关的高级服务功能。
"""

import asyncio
import os
import json
import math
import tempfile
import shutil
from typing import Dict, List, Any, Optional, Tuple
from pathlib import Path
from datetime import timedelta
import logging

from src.core.di import Inject, Service
from src.modules.processors import FFmpegProcessor

try:
    from ffmpeg import FFmpeg
    from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg

    FFMPEG_PYTHON_AVAILABLE = True
except ImportError:
    FFMPEG_PYTHON_AVAILABLE = False
    FFmpeg = None
    AsyncFFmpeg = None


@Service("ffmpeg_service")
class FFmpegService:
    """
    FFmpeg服务

    提供高级的FFmpeg操作和批处理功能。
    """

    def __init__(
        self,
        processor: FFmpegProcessor = Inject(FFmpegProcessor),
        config: Dict[str, Any] = Inject("config"),
        logger: logging.Logger = Inject("logger"),
    ):
        self.processor = processor
        self.config = config
        self.logger = logger

        # 批处理配置
        self.max_concurrent_jobs = config.get("ffmpeg", {}).get(
            "max_concurrent_jobs", 3
        )
        self.job_queue = asyncio.Queue()
        self.active_jobs = {}
        self.job_counter = 0

        # 检查python-ffmpeg可用性
        if not FFMPEG_PYTHON_AVAILABLE:
            self.logger.warning(
                "python-ffmpeg not available, some advanced features will be disabled"
            )

    def _create_ffmpeg_cmd(
        self, use_ffprobe: bool = False, quiet: bool = False
    ) -> Optional[FFmpeg]:
        """创建FFmpeg命令对象"""
        if not FFMPEG_PYTHON_AVAILABLE:
            return None

        if use_ffprobe:
            ffmpeg_cmd = FFmpeg("ffprobe")
        else:
            ffmpeg_cmd = FFmpeg().option("y").option("hide_banner")

        @ffmpeg_cmd.on("start")
        def on_start(arguments: list[str]):
            try:
                filter_index = arguments.index("-filter_complex")
                filter_content = arguments[filter_index + 1]
                arguments[filter_index + 1] = f'"{filter_content}"'
                args = " ".join(arguments)
                self.logger.info(f"FFmpeg command: {args}")
                arguments[filter_index + 1] = filter_content
            except ValueError:
                args = " ".join(arguments)
                self.logger.info(f"FFmpeg command: {args}")

        @ffmpeg_cmd.on("progress")
        def on_progress(progress):
            if not quiet:
                self.logger.info(f"处理进度: {progress}")

        @ffmpeg_cmd.on("completed")
        def on_completed():
            self.logger.info(f"FFmpeg task completed.")

        @ffmpeg_cmd.on("stderr")
        def on_stderr(line: str):
            if line.startswith("Error"):
                self.logger.error(line)
                raise RuntimeError(line)
            else:
                if not quiet:
                    self.logger.warning(line)

        return ffmpeg_cmd

    def _create_async_ffmpeg_cmd(
        self, use_ffprobe: bool = False, quiet: bool = False
    ) -> Optional[AsyncFFmpeg]:
        """创建异步FFmpeg命令对象"""
        if not FFMPEG_PYTHON_AVAILABLE:
            return None

        if use_ffprobe:
            ffmpeg_cmd = AsyncFFmpeg("ffprobe")
        else:
            ffmpeg_cmd = AsyncFFmpeg().option("y").option("hide_banner")

        @ffmpeg_cmd.on("start")
        def on_start(arguments: list[str]):
            try:
                filter_index = arguments.index("-filter_complex")
                filter_content = arguments[filter_index + 1]
                arguments[filter_index + 1] = f'"{filter_content}"'
                args = " ".join(arguments)
                self.logger.info(f"FFmpeg command: {args}")
                arguments[filter_index + 1] = filter_content
            except ValueError:
                args = " ".join(arguments)
                self.logger.info(f"FFmpeg command: {args}")

        @ffmpeg_cmd.on("progress")
        def on_progress(progress):
            if not quiet:
                self.logger.info(f"处理进度: {progress}")

        @ffmpeg_cmd.on("completed")
        def on_completed():
            self.logger.info(f"FFmpeg task completed.")

        @ffmpeg_cmd.on("stderr")
        def on_stderr(line: str):
            if line.startswith("Error") and ".m3u8" not in line:
                self.logger.error(line)
                raise RuntimeError(line)
            elif "Output file is empty" in line:
                raise RuntimeError("输出是空文件")
            else:
                if not quiet:
                    if "Skip" not in line:
                        self.logger.warning(line)

        return ffmpeg_cmd

    async def get_video_metadata(self, media_path: str) -> Dict[str, Any]:
        """
        获取视频元数据

        Args:
            media_path: 视频文件路径

        Returns:
            视频元数据字典
        """
        if not FFMPEG_PYTHON_AVAILABLE:
            # 回退到原有方法
            return await self.processor.get_video_info(media_path)

        ffprobe = self._create_ffmpeg_cmd(use_ffprobe=True)
        if not ffprobe:
            return await self.processor.get_video_info(media_path)

        ffprobe.input(
            media_path, print_format="json", show_streams=None, show_format=None
        )

        try:
            metadata_bytes = ffprobe.execute()
            metadata_json = json.loads(metadata_bytes)
            self.logger.debug(
                f"Video metadata: {json.dumps(metadata_json, ensure_ascii=False)}"
            )
            return metadata_json
        except Exception as e:
            self.logger.error(f"Failed to get video metadata: {e}")
            # 回退到原有方法
            return await self.processor.get_video_info(media_path)

    async def get_video_duration(self, media_path: str) -> float:
        """
        获取视频时长

        Args:
            media_path: 视频文件路径

        Returns:
            视频时长（秒）
        """
        metadata = await self.get_video_metadata(media_path)

        # 尝试从format中获取时长
        if "format" in metadata and "duration" in metadata["format"]:
            return float(metadata["format"]["duration"])

        # 尝试从streams中获取时长
        if "streams" in metadata:
            for stream in metadata["streams"]:
                if stream.get("codec_type") == "video" and "duration" in stream:
                    return float(stream["duration"])

        return 0.0

    async def get_video_size(self, media_path: str) -> Tuple[int, int]:
        """
        获取视频分辨率

        Args:
            media_path: 视频文件路径

        Returns:
            (width, height) 元组
        """
        metadata = await self.get_video_metadata(media_path)

        if "streams" in metadata:
            for stream in metadata["streams"]:
                if stream.get("codec_type") == "video":
                    width = stream.get("width", 0)
                    height = stream.get("height", 0)
                    return width, height

        return 0, 0

    async def slice_video_by_time(
        self,
        media_path: str,
        segments: List[Tuple[float, float]],
        output_dir: str,
        quality: str = "medium",
    ) -> List[str]:
        """
        按时间段切割视频

        Args:
            media_path: 输入视频路径
            segments: 时间段列表 [(start1, end1), (start2, end2), ...]
            output_dir: 输出目录
            quality: 输出质量

        Returns:
            输出文件路径列表
        """
        if not FFMPEG_PYTHON_AVAILABLE:
            # 回退到原有方法
            output_files = []
            for i, (start, end) in enumerate(segments):
                output_path = os.path.join(output_dir, f"segment_{i:03d}.mp4")
                await self.processor._extract_segment(
                    input_path=media_path,
                    output_path=output_path,
                    start_time=start,
                    end_time=end,
                    quality=quality,
                )
                output_files.append(output_path)
            return output_files

        os.makedirs(output_dir, exist_ok=True)

        ffmpeg_cmd = self._create_async_ffmpeg_cmd()
        if not ffmpeg_cmd:
            return []

        ffmpeg_cmd.input(media_path)

        filter_complex = []
        output_files = []

        # 质量设置
        quality_settings = {
            "low": {"crf": 28, "preset": "fast"},
            "medium": {"crf": 23, "preset": "medium"},
            "high": {"crf": 18, "preset": "slow"},
        }
        settings = quality_settings.get(quality, quality_settings["medium"])

        for index, (start, end) in enumerate(segments):
            # 创建过滤器
            filter_complex.extend(
                [
                    f"[v:0]trim=start={start}:end={end},setpts=PTS-STARTPTS[cut{index}]",
                    f"[a:0]atrim=start={start}:end={end},asetpts=PTS-STARTPTS[acut{index}]",
                ]
            )

            # 输出文件
            output_path = os.path.join(output_dir, f"segment_{index:03d}.mp4")
            output_files.append(output_path)

            ffmpeg_options = {
                "map": [f"[cut{index}]", f"[acut{index}]"],
                "reset_timestamps": "1",
                "sc_threshold": "0",
                "g": "1",
                "force_key_frames": "expr:gte(t,n_forced*1)",
                "vcodec": "libx264",
                "acodec": "aac",
                "crf": settings["crf"],
                "preset": settings["preset"],
            }

            ffmpeg_cmd.output(output_path, options=ffmpeg_options)

        if filter_complex:
            ffmpeg_cmd.option("filter_complex", ";".join(filter_complex))

        try:
            await ffmpeg_cmd.execute()
            self.logger.info(
                f"Successfully sliced video into {len(output_files)} segments"
            )
            return output_files
        except Exception as e:
            self.logger.error(f"Failed to slice video: {e}")
            raise

    async def concat_videos(
        self,
        video_files: List[str],
        output_path: str,
        target_width: int = 1080,
        target_height: int = 1920,
    ) -> str:
        """
        合并多个视频文件

        Args:
            video_files: 视频文件列表
            output_path: 输出文件路径
            target_width: 目标宽度
            target_height: 目标高度

        Returns:
            输出文件路径
        """
        if not video_files:
            raise ValueError("没有可以合并的视频源")

        if not FFMPEG_PYTHON_AVAILABLE:
            # 回退到原有方法
            await self._concat_videos(video_files, output_path)
            return output_path

        os.makedirs(os.path.dirname(output_path), exist_ok=True)

        ffmpeg_cmd = self._create_async_ffmpeg_cmd()
        if not ffmpeg_cmd:
            await self._concat_videos(video_files, output_path)
            return output_path

        filter_complex = []
        total_videos = len(video_files)

        # 添加所有输入文件
        for input_path in video_files:
            ffmpeg_cmd.input(input_path)

        # 统一所有视频的格式、分辨率和帧率
        for i in range(total_videos):
            filter_complex.extend(
                [
                    # 先缩放到统一分辨率，然后设置帧率和格式
                    f"[{i}:v]scale={target_width}:{target_height}:force_original_aspect_ratio=decrease,"
                    f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2,"
                    f"setsar=1:1,"  # 强制设置SAR
                    f"fps=30,format=yuv420p[v{i}]",
                    # 音频格式统一
                    f"[{i}:a]aformat=sample_fmts=s16:sample_rates=44100:channel_layouts=stereo[a{i}]",
                ]
            )

        # 准备处理后的视频流和音频流的连接字符串
        video_streams = "".join(f"[v{i}]" for i in range(total_videos))
        audio_streams = "".join(f"[a{i}]" for i in range(total_videos))

        # 使用concat过滤器合并视频和音频
        filter_complex.extend(
            [
                f"{video_streams}concat=n={total_videos}:v=1:a=0[vconcated]",
                f"{audio_streams}concat=n={total_videos}:v=0:a=1[aconcated]",
            ]
        )

        ffmpeg_cmd.output(
            output_path,
            {
                "filter_complex": ";".join(filter_complex),
                "map": ["[vconcated]", "[aconcated]"],
                "vcodec": "libx264",
                "crf": 16,
                "r": 30,
                "acodec": "aac",
                "ar": 44100,
                "ac": 2,
                "ab": "192k",
            },
        )

        try:
            await ffmpeg_cmd.execute()
            self.logger.info(f"Successfully concatenated {len(video_files)} videos")
            return output_path
        except Exception as e:
            self.logger.error(f"Failed to concatenate videos: {e}")
            raise

    async def extract_audio(
        self, media_path: str, output_path: Optional[str] = None
    ) -> str:
        """
        提取视频音频

        Args:
            media_path: 输入视频路径
            output_path: 输出音频路径

        Returns:
            输出音频文件路径
        """
        if not output_path:
            output_path = str(Path(media_path).with_suffix(".wav"))

        os.makedirs(os.path.dirname(output_path), exist_ok=True)

        if not FFMPEG_PYTHON_AVAILABLE:
            # 使用原有方法
            cmd = [
                self.processor.ffmpeg_path,
                "-i",
                media_path,
                "-map",
                "0:a",
                "-acodec",
                "pcm_s16le",
                "-ar",
                "44100",
                "-ac",
                "1",
                "-y",
                output_path,
            ]

            process = await asyncio.create_subprocess_exec(
                *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )

            stdout, stderr = await process.communicate()

            if process.returncode != 0:
                error_msg = stderr.decode() if stderr else "未知错误"
                raise RuntimeError(f"音频提取失败: {error_msg}")

            return output_path

        # 首先检查是否有音频流
        ffprobe_cmd = self._create_async_ffmpeg_cmd(use_ffprobe=True)
        if ffprobe_cmd:
            ffprobe_cmd.input(
                media_path,
                v="quiet",
                print_format="json",
                select_streams="a",
                show_entries="stream=codec_type",
            )

            try:
                audio_check_bytes = await ffprobe_cmd.execute()
                audio_check = json.loads(audio_check_bytes)
                self.logger.debug(f"Audio check: {audio_check}")

                if len(audio_check["streams"]) == 0:
                    raise RuntimeError(f"Media has no audio streams.")
            except Exception as e:
                self.logger.warning(f"Audio check failed: {e}")

        # 提取音频
        ffmpeg_cmd = self._create_async_ffmpeg_cmd()
        if not ffmpeg_cmd:
            raise RuntimeError("FFmpeg not available")

        ffmpeg_cmd.input(media_path).output(
            output_path, map="0:a", acodec="pcm_s16le", ar=44100, ac=1
        )

        try:
            await ffmpeg_cmd.execute()
            self.logger.info(f"Successfully extracted audio to {output_path}")
            return output_path
        except Exception as e:
            self.logger.error(f"Failed to extract audio: {e}")
            raise

    async def add_video_effects(
        self, media_path: str, output_path: str, effects: Dict[str, Any]
    ) -> str:
        """
        添加视频特效

        Args:
            media_path: 输入视频路径
            output_path: 输出视频路径
            effects: 特效配置字典

        Returns:
            输出视频路径
        """
        if not FFMPEG_PYTHON_AVAILABLE:
            raise RuntimeError("Advanced video effects require python-ffmpeg")

        os.makedirs(os.path.dirname(output_path), exist_ok=True)

        ffmpeg_cmd = self._create_async_ffmpeg_cmd()
        if not ffmpeg_cmd:
            raise RuntimeError("FFmpeg not available")

        ffmpeg_cmd.input(media_path)

        # 获取视频信息
        metadata = await self.get_video_metadata(media_path)
        video_stream = None
        for stream in metadata.get("streams", []):
            if stream.get("codec_type") == "video":
                video_stream = stream
                break

        if not video_stream:
            raise RuntimeError("No video stream found")

        width = video_stream.get("width", 1920)
        height = video_stream.get("height", 1080)

        video_filters = []

        # 缩放特效
        if "zoom" in effects:
            zoom_config = effects["zoom"]
            duration = (
                zoom_config.get("duration", 6.0) * 2
            )  # abs(sin())表达式会导致实际的往复频率为2倍
            zoom_factor = zoom_config.get("factor", 0.1)

            video_filters.append(
                f"scale={4 * width}x{4 * height},fps=30,"
                f"zoompan=z='1+{zoom_factor}*abs(sin(2*PI*time/{duration}))':"
                "x='trunc(iw/2*(1-1/zoom))':"
                "y='trunc(ih/2*(1-1/zoom))':"
                f"d=1:s={width}x{height}:fps=30"
            )

        # 镜像特效
        if "mirror" in effects:
            mirror_config = effects["mirror"]
            scale_down = mirror_config.get("scale_down", 6)
            from_right = mirror_config.get("from_right", True)
            position = mirror_config.get("position", (40, 40))

            mirror_x = (
                f"main_w-overlay_w-{position[0]}" if from_right else str(position[0])
            )

            filter_complex = [
                "[0:v]split[original][mirror]",
                f"[mirror]hflip,scale=iw/{scale_down}:-1,format=rgba[flipped]",
                "[flipped]split[fm1][fm2]",
                f"[fm2]format=gray,geq=lum='255*(1-pow(min(1,2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2))),1.5))':a='if(lt(pow(X/W-0.5,2)+pow(Y/H-0.5,2),0.15),(1-pow(2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2)),1.5))*255,0)'[fm2Blur]",
                "[fm1][fm2Blur]alphamerge[flipped_blured]",
                f"[original][flipped_blured]overlay=x={mirror_x}:y=main_h-overlay_h-{position[1]}[video]",
            ]

            ffmpeg_cmd.output(
                output_path,
                options={"filter_complex": ";".join(filter_complex)},
                map=["[video]", "0:a"],
                vcodec="libx264",
                crf=16,
            )
        else:
            # 应用视频滤镜
            if video_filters:
                ffmpeg_cmd.output(
                    output_path,
                    options={"vf": ",".join(video_filters)},
                    vcodec="libx264",
                    acodec="copy",
                    crf=16,
                )
            else:
                # 没有特效，直接复制
                ffmpeg_cmd.output(output_path, vcodec="copy", acodec="copy")

        try:
            await ffmpeg_cmd.execute()
            self.logger.info(f"Successfully applied effects to video: {output_path}")
            return output_path
        except Exception as e:
            self.logger.error(f"Failed to apply video effects: {e}")
            raise

    async def extract_frame(
        self, video_path: str, frame_time: float, output_path: Optional[str] = None
    ) -> str:
        """
        提取视频帧为图片

        Args:
            video_path: 视频文件路径
            frame_time: 提取帧的时间点（秒）
            output_path: 输出图片路径

        Returns:
            输出图片路径
        """
        if not output_path:
            output_path = str(Path(video_path).with_suffix(".jpg"))

        os.makedirs(os.path.dirname(output_path), exist_ok=True)

        if not FFMPEG_PYTHON_AVAILABLE:
            # 使用原有方法
            cmd = [
                self.processor.ffmpeg_path,
                "-i",
                video_path,
                "-ss",
                str(frame_time),
                "-vframes",
                "1",
                "-y",
                output_path,
            ]

            process = await asyncio.create_subprocess_exec(
                *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )

            _, stderr = await process.communicate()

            if process.returncode != 0:
                error_msg = stderr.decode() if stderr else "未知错误"
                raise RuntimeError(f"帧提取失败: {error_msg}")

            return output_path

        ffmpeg_cmd = self._create_async_ffmpeg_cmd()
        if not ffmpeg_cmd:
            raise RuntimeError("FFmpeg not available")

        ffmpeg_cmd.input(video_path, ss=frame_time)
        ffmpeg_cmd.output(output_path, vframes=1)

        try:
            await ffmpeg_cmd.execute()
            self.logger.info(
                f"Successfully extracted frame at {frame_time}s to {output_path}"
            )
            return output_path
        except Exception as e:
            self.logger.error(f"Failed to extract frame: {e}")
            raise

    async def batch_slice_videos(
        self,
        input_files: List[str],
        output_base_dir: str,
        max_duration: int = 30,
        min_duration: int = 3,
        quality: str = "medium",
        scene_detection: bool = True,
    ) -> Dict[str, Any]:
        """
        批量切割视频

        Args:
            input_files: 输入视频文件列表
            output_base_dir: 输出基础目录
            max_duration: 最大切片时长
            min_duration: 最小切片时长
            quality: 输出质量

        Returns:
            批处理结果
        """
        self.logger.info(f"开始批量切割 {len(input_files)} 个视频文件")

        # 创建输出目录
        os.makedirs(output_base_dir, exist_ok=True)

        # 创建任务
        tasks = []
        for input_file in input_files:
            file_name = Path(input_file).stem
            output_dir = os.path.join(output_base_dir, file_name)

            task = self._create_slice_task(
                input_file,
                output_dir,
                max_duration,
                min_duration,
                quality,
                scene_detection,
            )
            tasks.append(task)

        # 并发执行任务
        semaphore = asyncio.Semaphore(self.max_concurrent_jobs)

        async def limited_task(task):
            async with semaphore:
                return await task

        results = await asyncio.gather(
            *[limited_task(task) for task in tasks], return_exceptions=True
        )

        # 整理结果
        success_count = 0
        error_count = 0
        all_slices = []
        errors = []

        for i, result in enumerate(results):
            if isinstance(result, Exception):
                error_count += 1
                errors.append({"file": input_files[i], "error": str(result)})
                self.logger.error(f"处理文件失败 {input_files[i]}: {result}")
            else:
                success_count += 1
                all_slices.extend(result)

        batch_result = {
            "total_files": len(input_files),
            "success_count": success_count,
            "error_count": error_count,
            "total_slices": len(all_slices),
            "slices": all_slices,
            "errors": errors,
        }

        self.logger.info(
            f"批量切割完成: 成功 {success_count}, 失败 {error_count}, 总切片 {len(all_slices)}"
        )
        return batch_result

    async def _create_slice_task(
        self,
        input_file: str,
        output_dir: str,
        max_duration: int,
        min_duration: int,
        quality: str,
        scene_detection: bool = True,
    ) -> List[str]:
        """创建切片任务"""
        try:
            return await self.processor.slice_video(
                input_path=input_file,
                output_dir=output_dir,
                max_duration=max_duration,
                min_duration=min_duration,
                quality=quality,
                scene_detection=scene_detection,
            )
        except Exception as e:
            self.logger.error(f"切片任务失败 {input_file}: {e}")
            raise

    async def analyze_video_batch(self, video_files: List[str]) -> Dict[str, Any]:
        """
        批量分析视频

        Args:
            video_files: 视频文件列表

        Returns:
            分析结果汇总
        """
        self.logger.info(f"开始批量分析 {len(video_files)} 个视频文件")

        # 并发获取视频信息
        semaphore = asyncio.Semaphore(self.max_concurrent_jobs)

        async def analyze_single(video_file):
            async with semaphore:
                try:
                    return await self.processor.get_video_info(video_file)
                except Exception as e:
                    self.logger.error(f"分析视频失败 {video_file}: {e}")
                    return {"error": str(e), "file": video_file}

        results = await asyncio.gather(
            *[analyze_single(video_file) for video_file in video_files]
        )

        # 统计分析结果
        total_duration = 0
        total_size = 0
        format_stats = {}
        codec_stats = {}
        resolution_stats = {}
        errors = []

        for i, info in enumerate(results):
            if "error" in info:
                errors.append(info)
                continue

            # 累计统计
            total_duration += info.get("duration", 0)
            total_size += info.get("size", 0)

            # 格式统计
            format_name = info.get("format_name", "unknown")
            format_stats[format_name] = format_stats.get(format_name, 0) + 1

            # 编码器统计
            video_codec = info.get("video_codec", "unknown")
            codec_stats[video_codec] = codec_stats.get(video_codec, 0) + 1

            # 分辨率统计
            width = info.get("width", 0)
            height = info.get("height", 0)
            resolution = f"{width}x{height}"
            resolution_stats[resolution] = resolution_stats.get(resolution, 0) + 1

        analysis_result = {
            "total_files": len(video_files),
            "analyzed_files": len(video_files) - len(errors),
            "error_count": len(errors),
            "total_duration": total_duration,
            "total_size": total_size,
            "average_duration": total_duration / max(1, len(video_files) - len(errors)),
            "format_distribution": format_stats,
            "codec_distribution": codec_stats,
            "resolution_distribution": resolution_stats,
            "errors": errors,
        }

        self.logger.info(
            f"批量分析完成: 分析 {analysis_result['analyzed_files']} 个文件"
        )
        return analysis_result

    async def create_video_preview(
        self,
        video_files: List[str],
        output_path: str,
        preview_duration: int = 5,
        quality: str = "medium",
    ) -> str:
        """
        创建视频预览

        Args:
            video_files: 视频文件列表
            output_path: 输出文件路径
            preview_duration: 每个视频的预览时长
            quality: 输出质量

        Returns:
            预览视频路径
        """
        self.logger.info(f"创建视频预览: {len(video_files)} 个文件")

        # 创建临时目录
        temp_dir = os.path.join(os.path.dirname(output_path), "temp_preview")
        os.makedirs(temp_dir, exist_ok=True)

        try:
            # 为每个视频创建预览片段
            preview_segments = []
            for i, video_file in enumerate(video_files):
                segment_path = os.path.join(temp_dir, f"preview_{i:03d}.mp4")

                # 提取前几秒作为预览
                await self.processor._extract_segment(
                    input_path=video_file,
                    output_path=segment_path,
                    start_time=0,
                    end_time=preview_duration,
                    quality=quality,
                )

                preview_segments.append(segment_path)

            # 合并预览片段
            await self._concat_videos(preview_segments, output_path)

            self.logger.info(f"视频预览创建完成: {output_path}")
            return output_path

        finally:
            # 清理临时文件
            import shutil

            if os.path.exists(temp_dir):
                shutil.rmtree(temp_dir)

    async def _concat_videos(self, video_files: List[str], output_path: str) -> None:
        """合并视频文件"""
        # 创建文件列表
        list_file = output_path + ".list"

        try:
            with open(list_file, "w") as f:
                for video_file in video_files:
                    f.write(f"file '{os.path.abspath(video_file)}'\n")

            # 使用FFmpeg合并
            cmd = [
                self.processor.ffmpeg_path,
                "-f",
                "concat",
                "-safe",
                "0",
                "-i",
                list_file,
                "-c",
                "copy",
                "-y",
                output_path,
            ]

            process = await asyncio.create_subprocess_exec(
                *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )

            stdout, stderr = await process.communicate()

            if process.returncode != 0:
                error_msg = stderr.decode() if stderr else "未知错误"
                raise RuntimeError(f"视频合并失败: {error_msg}")

        finally:
            # 清理列表文件
            if os.path.exists(list_file):
                os.remove(list_file)

    async def optimize_video_for_web(
        self, input_path: str, output_path: str, target_size_mb: Optional[int] = None
    ) -> Dict[str, Any]:
        """
        为Web优化视频

        Args:
            input_path: 输入视频路径
            output_path: 输出视频路径
            target_size_mb: 目标文件大小（MB）

        Returns:
            优化结果信息
        """
        self.logger.info(f"开始Web优化: {input_path}")

        # 获取原始视频信息
        original_info = await self.processor.get_video_info(input_path)
        original_size_mb = original_info.get("size", 0) / (1024 * 1024)

        # 计算目标比特率
        if target_size_mb:
            duration = original_info.get("duration", 1)
            target_bitrate = int((target_size_mb * 8 * 1024) / duration)  # kbps
        else:
            target_bitrate = 1000  # 默认1Mbps

        # Web优化参数
        cmd = [
            self.processor.ffmpeg_path,
            "-i",
            input_path,
            "-c:v",
            "libx264",
            "-preset",
            "medium",
            "-crf",
            "23",
            "-maxrate",
            f"{target_bitrate}k",
            "-bufsize",
            f"{target_bitrate * 2}k",
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            "-movflags",
            "+faststart",  # 优化Web播放
            "-pix_fmt",
            "yuv420p",
            "-y",
            output_path,
        ]

        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        if process.returncode != 0:
            error_msg = stderr.decode() if stderr else "未知错误"
            raise RuntimeError(f"Web优化失败: {error_msg}")

        # 获取优化后的信息
        optimized_info = await self.processor.get_video_info(output_path)
        optimized_size_mb = optimized_info.get("size", 0) / (1024 * 1024)

        result = {
            "original_size_mb": original_size_mb,
            "optimized_size_mb": optimized_size_mb,
            "compression_ratio": original_size_mb / max(optimized_size_mb, 0.001),
            "size_reduction_percent": (
                (original_size_mb - optimized_size_mb) / original_size_mb
            )
            * 100,
            "target_bitrate_kbps": target_bitrate,
            "original_info": original_info,
            "optimized_info": optimized_info,
        }

        self.logger.info(f"Web优化完成: 压缩比 {result['compression_ratio']:.2f}")
        return result

    def get_processing_stats(self) -> Dict[str, Any]:
        """获取处理统计信息"""
        return {
            "max_concurrent_jobs": self.max_concurrent_jobs,
            "active_jobs": len(self.active_jobs),
            "queue_size": self.job_queue.qsize(),
            "total_jobs_processed": self.job_counter,
        }
